package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

public class Flink04_Source_Custom {
    public static void main(String[] args) throws Exception {
        //1.流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.设置并行度为1
        env.setParallelism(1);

        //3.TODO 自定义source
        DataStreamSource<WaterSensor> streamSource = env.addSource(new MySource());

        streamSource.print();

        env.execute();

    }

    //如果要设置多并行度的话，则需要用以下方式，实现并行度接口
//    public static class MySource implements ParallelSourceFunction<WaterSensor> {


    public static class MySource implements SourceFunction<WaterSensor> {
        private Random random = new Random();
        private Boolean isRunning=true;

        /**
         * 生成数据的方法，通常是由while循环
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            while (isRunning){
                ctx.collect(new WaterSensor("sensor_"+random.nextInt(1000), System.currentTimeMillis(), random.nextInt(500)));
            }
        }

        /**
         * 取消数据源，通常是run方法中while循环的终止条件
         */
        @Override
        public void cancel() {

            isRunning = false;
        }
    }
}
